package bsf.baseservice.businessmq.db;

import bsf.baseservice.businessmq.dal.*;
import bsf.baseservice.businessmq.model.*;
import bsf.baseservice.businessmq.systemruntime.*;

import bsf.db.DbConn;
import org.apache.commons.lang3.StringUtils;

public class ConsumerBLL extends BaseBLL
{
	public String getDataNodeConnectString(DbConn PubConn, int datanodepartition)
	{
		tb_datanode_dal dal = new tb_datanode_dal();
		tb_datanode_model model = dal.get2(PubConn, datanodepartition);
		if (model == null)
		{
			throw new BusinessMQException("当前数据节点不存在:" + datanodepartition);
		}
		return getDataNodeConnectString(SystemParamConfig.Consumer_DataNode_ConnectString_Template, model);
	}

	public tb_consumer_client_model registerClient(DbConn PubConn, String client)
	{
		tb_consumer_client_dal dal = new tb_consumer_client_dal();
		tb_consumer_client_model clientmodel = dal.getByClient(PubConn, client);
		if (clientmodel==null)
		{
			dal.add2(PubConn, client);
			clientmodel = dal.getByClient(PubConn, client);
		}
		return clientmodel;
	}

	public tb_consumer_model registerConsumer(DbConn PubConn, int clientid, String clientname, java.util.ArrayList<Integer> partitionindexs)
	{
		long tempid = CommonHelper.generateIntID();
		tb_consumer_dal dal = new tb_consumer_dal();
		dal.deleteNotOnLineByClientID(PubConn, clientid, SystemParamConfig.Consumer_ConsumerHeartbeat_MAX_TIME_OUT);
		java.util.ArrayList<Integer> usedpartitionindexs = dal.getRegisterPartitionIndexs(PubConn, clientid);
		/*var conflictpartitionindexs = (from o in usedpartitionindexs from n in partitionindexs where o == n select o).ToList();*/
		java.util.ArrayList<Integer> conflictpartitionindexs = new java.util.ArrayList<Integer>();
		for(Integer o :usedpartitionindexs)
		{
			for (Integer n:partitionindexs)
			{
				if(o==n)
				{conflictpartitionindexs.add(o);}
			}
		}
		if (conflictpartitionindexs != null && conflictpartitionindexs.size() > 0)
		{
			throw new BusinessMQException(String.format("当前分区序号已经被注册使用中,冲突分区序号为:%1$s,可能是上次消费者异常终止导致消费者依然在注册中,请尝试在%2$s秒系统超时后重试。",
					StringUtils.join(conflictpartitionindexs.toArray(),","),
					SystemParamConfig.Consumer_ConsumerHeartbeat_MAX_TIME_OUT));
		}
		tb_consumer_model tempVar = new tb_consumer_model();
		tempVar.clientname=clientname;
		tempVar.consumerclientid=clientid;
		tempVar.partitionindexs=StringUtils.strip(StringUtils.join(partitionindexs.toArray(),','),",");
		tempVar.tempid=tempid;
		dal.add2(PubConn, tempVar);
		return dal.get(PubConn, tempid, clientid);
	}

	public  void removeConsumer(DbConn PubConn, long tempid, int clientid)
	{
		tb_consumer_dal dal = new tb_consumer_dal();
		dal.deleteClient(PubConn, clientid, tempid);

	}

	public  void consumerHeartbeat(DbConn PubConn, long tempid, int clientid)
	{
		tb_consumer_dal dal = new tb_consumer_dal();
		dal.clientHeatbeat(PubConn, clientid, tempid);
	}

	public tb_mqpath_model getMQPath(DbConn PubConn, String mqpath)
	{
		tb_mqpath_dal mqpathdal = new tb_mqpath_dal();
		tb_mqpath_model mqpathmodel = mqpathdal.get(PubConn, mqpath);
		if (mqpathmodel == null)
		{
			throw new BusinessMQException("当前队列不存在:" + mqpath);
		}
		return mqpathmodel;
	}

	public  java.util.Date getLastUpdateTimeOfMqPath(DbConn PubConn, String mqpath)
	{
		tb_mqpath_dal mqpathdal = new tb_mqpath_dal();
		return mqpathdal.getLastUpdateTimeOfMqPath(PubConn, mqpath);
	}

	public tb_consumer_partition_model registerConsumerPartition(DbConn PubConn, int clientid, int partitionindex, String mqpath, long tempid)
	{
		tb_mqpath_model mqpathmodel = getMQPath(PubConn, mqpath);
		tb_mqpath_partition_dal mqpathpartitiondal = new tb_mqpath_partition_dal();
		tb_mqpath_partition_model mqpathpartitionmodel = mqpathpartitiondal.getOfConsumer(PubConn, partitionindex, mqpathmodel.id);
		if (mqpathpartitionmodel == null)
		{
			return null;
		}
		tb_consumer_partition_dal dal = new tb_consumer_partition_dal();
		tb_consumer_partition_model tempVar = new tb_consumer_partition_model();
		tempVar.consumerclientid=clientid;
		tempVar.lastconsumertempid=tempid;
		tempVar.partitionid=mqpathpartitionmodel.partitionid;
		tempVar.partitionindex=partitionindex;
		if (dal.edit2(PubConn, tempVar) <= 0)
		{
			PartitionIDInfo partitionidinfo = PartitionRuleHelper.getPartitionIDInfo(mqpathpartitionmodel.partitionid);
			String datanodename = PartitionRuleHelper.getDataNodeName(partitionidinfo.DataNodePartition);
			long maxid = -1;
			java.util.Date serverdate = PubConn.getServerDate();
			String tablename = PartitionRuleHelper.getTableName(partitionidinfo.TablePartition, serverdate);
			try(DbConn c = DbConn.createConn(this.getDataNodeConnectString(PubConn, partitionidinfo.DataNodePartition)))
			{
				boolean exist = c.tableIsExist(tablename);
				if (!exist)
				{
					throw new BusinessMQException(String.format("当前数据节点%1$s,表%2$s不存在", partitionidinfo.DataNodePartition,tablename));
				}
				tb_messagequeue_dal mqdal = new tb_messagequeue_dal();
				mqdal.TableName=tablename;
				maxid = mqdal.getMaxId(c);
				if (maxid <= 0)
				{
					MQIDInfo tempVar2 = new MQIDInfo();
					tempVar2.AutoID=0;
					tempVar2.DataNodePartition=partitionidinfo.DataNodePartition;
					tempVar2.Day=serverdate;
					tempVar2.TablePartition=partitionidinfo.TablePartition;
					maxid = PartitionRuleHelper.getMQID(tempVar2);
				}
			}

			tb_consumer_partition_model tempVar3 = new tb_consumer_partition_model();
			tempVar3.consumerclientid=clientid;
			tempVar3.lastconsumertempid=tempid;
			tempVar3.lastmqid=maxid;
			tempVar3.partitionid=mqpathpartitionmodel.partitionid;
			tempVar3.partitionindex=partitionindex;
			dal.add2(PubConn, tempVar3);
		}
		return dal.get(PubConn, clientid, mqpathpartitionmodel.partitionid);
	}


}